package com.reactor.operator;

import java.time.Duration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.junit.Test;

import reactor.core.publisher.Flux;

//假设我们有一个关于zip方法的说明：“Zip two sources together, that is to say wait for all the sources to emit one element and combine these elements once into a Tuple2.”，我们希望将这句话拆分为一个一个的单词并以每200ms一个的速度发出，除了前面flatMap的例子中用到的delayElements，可以如下操作：
public class Zip {

	private Flux<String> getZipDescFlux() {
	    String desc = "Zip two sources together, that is to say wait for all the sources to emit one element and combine these elements once into a Tuple2.";
	    return Flux.fromArray(desc.split("\\s+"));  // 将英文说明用空格拆分为字符串流
	}

	@Test
	public void testSimpleOperators() throws InterruptedException {
	    CountDownLatch countDownLatch = new CountDownLatch(1);  // 定义一个CountDownLatch，初始为1，则会等待执行1次countDown方法后结束，不使用它的话，测试方法所在的线程会直接返回而不会等待数据流发出完毕
	    Flux.zip(getZipDescFlux(), Flux.interval(Duration.ofMillis(200)))  // 使用Flux.interval声明一个每200ms发出一个元素的long数据流；因为zip操作是一对一的，故而将其与字符串流zip之后，字符串流也将具有同样的速度
	        .subscribe(t -> System.out.println(t.getT1()), null, countDownLatch::countDown);    // zip之后的流中元素类型为Tuple2，使用getT1方法拿到字符串流的元素；定义完成信号的处理为countDown
	    countDownLatch.await(10, TimeUnit.SECONDS);     // countDownLatch.await(10, TimeUnit.SECONDS)会等待countDown倒数至0，最多等待10秒钟
	}

}
